package com.kun.goods.mq;

import com.alibaba.fastjson.JSON;
import com.kun.api.IGoodsNumberLogService;
import com.kun.api.IGoodsService;
import com.kun.common.constant.ShopCode;
import com.kun.goods.service.TradeMqConsumerLogService;
import com.kun.pojo.domain.TradeGoods;
import com.kun.pojo.domain.TradeGoodsNumberLog;
import com.kun.pojo.domain.TradeMqConsumerLog;
import com.kun.pojo.entity.MQEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Date;

@Component
@Slf4j
@RocketMQMessageListener(topic = "${mq.order.topic}", consumerGroup = "${mq.order.consumer.group}",
        messageModel = MessageModel.BROADCASTING)
public class CancelMQListener implements RocketMQListener<MessageExt> {

    @Value("${mq.order.consumer.group}")
    private String groupName;

    @Autowired
    private TradeMqConsumerLogService tradeMqConsumerLogService;

    @Autowired
    private IGoodsService goodsService;

    @Autowired
    private IGoodsNumberLogService goodsNumberLogService;

    @Override
    public void onMessage(MessageExt messageExt) {
        // 1. 解析消息内容
        String msgId = messageExt.getMsgId();
        String tags = messageExt.getTags();
        String keys = messageExt.getKeys();
        String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
        log.info("接受消息成功");
        try {
            // 2. 查询消息消费记录
            TradeMqConsumerLog mqConsumerLog = tradeMqConsumerLogService.lambdaQuery().eq(TradeMqConsumerLog::getMsgTag, tags)
                    .eq(TradeMqConsumerLog::getMsgKey, keys).eq(TradeMqConsumerLog::getGroupName, groupName)
                    .last("LIMIT 1").one();
            // 3. 判断如果消费过
            if (mqConsumerLog != null) {
                // 获得消息处理状态
                Integer consumerStatus = mqConsumerLog.getConsumerStatus();
                if (ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode().equals(consumerStatus)) {
                    log.info("消息：{}已经处理完成", msgId);
                    return;
                }
                // 正在处理
                if (ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode().equals(consumerStatus)) {
                    log.info("消息：{}正在处理", msgId);
                    return;
                }
                // 处理失败
                if (ShopCode.SHOP_MQ_MESSAGE_STATUS_FAIL.getCode().equals(consumerStatus)) {
                    Integer consumerTimes = mqConsumerLog.getConsumerTimes();
                    if (consumerTimes > 3) {
                        log.info("消息：{}已经处理超过3次，不能进行消费了。", msgId);
                        return;
                    }
                    boolean flag = tradeMqConsumerLogService.lambdaUpdate()
                            .set(TradeMqConsumerLog::getConsumerStatus, ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode())
                            .eq(TradeMqConsumerLog::getMsgTag, tags)
                            .eq(TradeMqConsumerLog::getMsgKey, keys)
                            .eq(TradeMqConsumerLog::getGroupName, groupName)
                            .eq(TradeMqConsumerLog::getConsumerTimes, consumerTimes).update();
                    if (!flag) {
                        log.info("并发修改，请稍后处理。");
                    }
                }
            } else {
                // 4. 判断没有消费过
                TradeMqConsumerLog tradeMqConsumerLog = new TradeMqConsumerLog();
                tradeMqConsumerLog.setMsgTag(tags);
                tradeMqConsumerLog.setMsgKey(keys);
                tradeMqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());
                tradeMqConsumerLog.setMsgBody(body);
                tradeMqConsumerLog.setMsgId(msgId);
                tradeMqConsumerLog.setConsumerTimes(0);
                tradeMqConsumerLog.setGroupName(groupName);
                // 将信息添加到数据库
                tradeMqConsumerLogService.save(tradeMqConsumerLog);
            }
            // 5. 回退库存
            MQEntity entity = JSON.parseObject(body, MQEntity.class);
            // 商品ID
            TradeGoods tradeGoods = goodsService.findOne(entity.getGoodsId());
            goodsService.lambdaUpdate()
                    .set(TradeGoods::getGoodsNumber, tradeGoods.getGoodsNumber() + entity.getGoodsNum())
                    .eq(TradeGoods::getGoodsId, tradeGoods.getGoodsId()).update();
            // 记录操作日志
            goodsNumberLogService.save(new TradeGoodsNumberLog(entity.getGoodsId(), entity.getOrderId(),
                    entity.getGoodsNum(), new Date()));
            // 6. 记录消息状态为处理成功
            tradeMqConsumerLogService.lambdaUpdate()
                    .set(TradeMqConsumerLog::getConsumerStatus, ShopCode.SHOP_MQ_MESSAGE_STATUS_SUCCESS.getCode())
                    .eq(TradeMqConsumerLog::getMsgTag, tags)
                    .eq(TradeMqConsumerLog::getMsgKey, keys)
                    .eq(TradeMqConsumerLog::getGroupName, groupName).update();
            log.info("回退库存成功");
        } catch (Exception e) {
            TradeMqConsumerLog mqConsumerLog = tradeMqConsumerLogService.lambdaQuery().eq(TradeMqConsumerLog::getMsgTag, tags)
                    .eq(TradeMqConsumerLog::getMsgKey, keys).eq(TradeMqConsumerLog::getGroupName, groupName)
                    .last("LIMIT 1").one();
            // 对错误进行更新
            if (mqConsumerLog == null) {
                mqConsumerLog = new TradeMqConsumerLog();
                mqConsumerLog.setMsgTag(tags);
                mqConsumerLog.setMsgKey(keys);
                mqConsumerLog.setConsumerStatus(ShopCode.SHOP_MQ_MESSAGE_STATUS_PROCESSING.getCode());
                mqConsumerLog.setMsgBody(body);
                mqConsumerLog.setMsgId(msgId);
                mqConsumerLog.setConsumerTimes(1);
                mqConsumerLog.setGroupName(groupName);
                tradeMqConsumerLogService.save(mqConsumerLog);
            } else {
                tradeMqConsumerLogService.lambdaUpdate()
                        .set(TradeMqConsumerLog::getConsumerTimes, mqConsumerLog.getConsumerTimes() + 1)
                        .eq(TradeMqConsumerLog::getMsgTag, tags)
                        .eq(TradeMqConsumerLog::getMsgKey, keys)
                        .eq(TradeMqConsumerLog::getGroupName, groupName).update();
            }
        }

    }
}
